package com.example.transform;

import com.example.beans.SensorReading;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import sun.awt.SunHints;

import javax.xml.crypto.Data;


public class TransformTest2_RollingAggregation {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 从文件读取数据
        DataStream<String> inputStream = env.readTextFile("E:\\hjx\\ideapro\\myflinkstu\\src\\main\\resources\\sensor.txt");

        // 转换成SensorReading类型
//        DataStream<SensorReading> dataStream = inputStream.map(new MapFunction<String, SensorReading>() {
//            @Override
//            public SensorReading map(String value) throws Exception {
//                String[] fields = value.split(",");
//                return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
//            }
//        });

        DataStream<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        } );

        // 分组,以下4条等价
        KeyedStream<SensorReading, Tuple> keyedStream = dataStream.keyBy("id");
        KeyedStream<SensorReading, String> keyedStream1 = dataStream.keyBy(data -> data.getId());
//        KeyedStream<SensorReading, String> keyedStream__ = dataStream.keyBy(SensorReading::getId);
//        KeyedStream<SensorReading, String> keyedStream_ = dataStream.keyBy(new KeySelector<SensorReading, String>() {
//            @Override
//            public String getKey(SensorReading value) throws Exception {
//                return value.getId();
//            }
//        });

        DataStream<Long> dataStream1 = env.fromElements(1L, 34L, 4L, 657L, 23L);
        KeyedStream<Long, Integer> keyedStream2 = dataStream1.keyBy(new KeySelector<Long, Integer>() {
            @Override
            public Integer getKey(Long value) throws Exception {
                return value.intValue() % 2;
            }
        });

        //        KeyedStream<SensorReading, String> keyedStream1 = dataStream.keyBy(SensorReading::getId);

        // 滚动聚合，取当前最大的温度值
        // 还有这些函数 min max minBy maxBy sum reduce
        // maxBy 获取当前最大值及其对应的所有字段
        DataStream<SensorReading> resultStream = keyedStream.maxBy("temperature");

        /*
        reduce:
        一个分组数据流的聚合操作，合并当前的元素
        和上次聚合的结果，产生一个新的值，返回的流中包含每一次聚合的结果，而不是
        只返回最后一次聚合的最终结果。
         */
        DataStream<SensorReading> reduce = keyedStream.reduce(new ReduceFunction<SensorReading>() {
            @Override
            public SensorReading reduce(SensorReading sensorReading, SensorReading t1) throws Exception {
                return new SensorReading(
                        sensorReading.getId(),
                        sensorReading.getTimestamp(),
                        Math.max(sensorReading.getTemperature(), t1.getTemperature())
                );
            }
        });

//        resultStream.print("result");

        reduce.print("reduce");

//        keyedStream1.print("key1");
//        keyedStream2.sum(0).print("key2");
        env.execute();
    }
}
